package cn.com.duiba.millionaire.center.biz.service.impl;

import cn.com.duiba.millionaire.center.api.constant.RedisKeyFactory;
import cn.com.duiba.millionaire.center.biz.service.WinnerListService;
import org.apache.commons.collections.CollectionUtils;
import org.javatuples.Pair;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

/**
 * Created by zzy on 2018/1/20.
 */
@Service
public class WinnerListServiceImpl implements WinnerListService {
    /**
     * key:活动id,Pair中key:用户id,value:应用id
     */
    @Resource(name = "redisTemplate")
    private RedisTemplate<String, Pair<Long, Long>> redisTemplate;

    @Override
    public void batchAdd(Map<Long, Set<Pair<Long, Long>>> winners) {

        redisTemplate.executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                RedisSerializer<Object> keyS = (RedisSerializer<Object>) redisTemplate.getKeySerializer();
                RedisSerializer<Object> valueS = (RedisSerializer<Object>) redisTemplate.getValueSerializer();
                for (Map.Entry<Long, Set<Pair<Long, Long>>> entry : winners.entrySet()) {
                    byte[] keyBytes = keyS.serialize(buildKey(entry.getKey()));
                    for (Pair<Long, Long> win : entry.getValue()) {
                        connection.sAdd(keyBytes, valueS.serialize(win));
                    }
                    connection.expire(keyBytes, 3 * 24 * 3600);
                }
                return null;
            }
        });
    }

    @Override
    public Set<Pair<Long, Long>> batchFindCids(Long activityId, int count) {
        List<Pair<Long, Long>> ids = redisTemplate.opsForSet().randomMembers(buildKey(activityId), count);
        if (CollectionUtils.isEmpty(ids)) {
            return Collections.emptySet();
        }
        return new HashSet<>(ids);
    }

    @Override
    public Long countWinners(Long activityId) {
        return redisTemplate.opsForSet().size(buildKey(activityId));
    }

    @Override
    public Set<Pair<Long, Long>> findAllWinners(Long activityId) {
        Set<Pair<Long, Long>> ids = redisTemplate.opsForSet().members(buildKey(activityId));
        if (CollectionUtils.isEmpty(ids)) {
            return Collections.emptySet();
        }
        return ids;
    }

    private String buildKey(Long activityId) {
        return RedisKeyFactory.K003.toString() + activityId;
    }
}
